package com.doit.log_collect.flume;

import com.google.gson.Gson;
import com.google.gson.JsonSyntaxException;
import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.interceptor.Interceptor;

import java.util.List;

/**
 * @DATE 2022/3/9/16:14
 * @Author MDK
 * @Version 2021.2.2
 **/
public class LogTimeExtract implements Interceptor {

    Gson gson = null;
    String keyName;

    public LogTimeExtract(String keyName) {
        this.keyName = keyName;
    }

    @Override
    public void initialize() {
        gson = new Gson();
    }

    @Override
    public Event intercept(Event event) {
        //如果获取到的json串信息缺失,程序不能终止
        try {
            //获取日志数据并解析
            String log = new String(event.getBody());
            long timestamp = gson.fromJson(log, LogBean.class).getTimestamp();
            event.getHeaders().put(keyName, timestamp+"");
            return event;
        } catch (JsonSyntaxException e) {
            e.printStackTrace();
            return null;
        }
    }

    @Override
    public List<Event> intercept(List<Event> list) {
        //按照批次处理event,调用上面的intercept重载方法
        for (Event event : list) {
            intercept(event);
        }
        return list;
    }

    @Override
    public void close() {
        gson =  null;
    }

    //提供一个拦截器类的对象构造器
    public static class ExtractBuilder implements Interceptor.Builder{

        String keyName = "timestamp";

        @Override
        public Interceptor build() {
            return new LogTimeExtract(keyName);
        }

        //configure方法可以获取对于source,channel,sink的配置信息
        @Override
        public void configure(Context context) {
            keyName = context.getString("keyname");
        }
    }

}
